package operator;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.util.Collector;

/**
 * connect 算子
 */
public class ConnectDemo2 {

    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        DataStreamSource<String> lines1 = env.socketTextStream("localhost", 8888);
        DataStreamSource<String> lines2 = env.socketTextStream("localhost", 9999);
        SingleOutputStreamOperator<Integer> nums = lines2.map(Integer::parseInt);

        // 连接两个Stream
        ConnectedStreams<String, Integer> connect = lines1.connect(nums);

        SingleOutputStreamOperator<String> maped = connect.flatMap(new CoFlatMapFunction<String, Integer, String>() {
            @Override
            public void flatMap1(String s, Collector<String> collector) throws Exception {
                String[] words = s.split(",");
                for (String word : words) {
                    collector.collect(word);
                }
            }

            @Override
            public void flatMap2(Integer integer, Collector<String> collector) throws Exception {
                collector.collect(integer.toString());
            }
        });

        maped.print();
        env.execute();
    }
}
